package com.ruoyi.system.listener;

import com.alibaba.fastjson2.JSON;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.system.domain.resp.SysContentResp;
import com.ruoyi.system.domain.resp.SysTalkBodyResp;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.FluxSink;

/**
 * 订阅类
 */
@Slf4j
public class SparkAISubscriber implements Subscriber<String>, Disposable {

    // 订阅对象
    private Subscription subscription;

    // emitter: 用于发射数据给订阅者
    private final FluxSink<String> emitter;

    // 用于数据的累积
    private final StringBuilder sb;

    //回调方法
    private final CompletedCallBack completedCallBack;

    public SparkAISubscriber(FluxSink<String> emitter, CompletedCallBack completedCallBack) {
        this.emitter = emitter;
        this.completedCallBack = completedCallBack;
        this.sb = new StringBuilder(2);
    }


    /**
     * onSubscribe: 当订阅建立时被调用，请求一个数据项。
     * onNext: 当接收到新数据时被调用，根据消息类型处理数据，将处理结果发送给订阅者，并进行相应的数据累积。
     * onError: 当遇到错误时被调用，向订阅者发送错误信息，并执行相应的清理操作。
     * onComplete: 当订阅完成时被调用，根据消息类型最终处理数据并发送完成信号。
     */
    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        // 请求下一个数据项
        subscription.request(1);
    }

    @Override
    public void onNext(String data) {
        // 处理每个元素的逻辑
//        log.info("SparkAI返回数据：{}", data);
        SysContentResp res = SysContentResp.builder()
                .end(Boolean.FALSE)
                .build();

        if ("[DONE]".equals(data)) {
//            log.info("SparkAI返回数据结束了");
            subscription.request(1);
            res.setEnd(Boolean.TRUE);
            emitter.next(JSON.toJSONString(AjaxResult.success(res)));
            emitter.complete();
        } else {
            SysTalkBodyResp sysTalkBodyResp = JSON.parseObject(data, SysTalkBodyResp.class);
            String content = sysTalkBodyResp.getChoices().get(0).getDelta().getContent();
            content = content == null ? "" : content;
            emitter.next(JSON.toJSONString(AjaxResult.success(content)));
            sb.append(content);
            // 请求下一个数据项
            subscription.request(1);
        }
    }

    @Override
    public void onError(Throwable t) {
        // 处理错误的逻辑
        emitter.next(JSON.toJSONString(AjaxResult.error(t.getMessage())));
        emitter.complete();
    }

    @Override
    public void onComplete() {
        // 数据流完成的处理
        emitter.complete();
    }

    @Override
    public void dispose() {
        log.warn("SparkAI返回数据关闭");
        if (subscription != null) {
            subscription.cancel(); // 取消订阅
            // 进行其他资源的释放工作
        }
        emitter.complete();
    }
}
